#ifndef __M_MSG_H__
#define __M_MSG_H__

/**
 * @file message.hpp
 * @brief 消息管理模块
 * 
 * 本模块实现消息队列绑定关系的声明、删除、持久化存储及内存管理功能。
 * 包含三个核心组件：
 * 1. MessageMapper - 负责消息的持久化存储管理，处理文件读写、垃圾回收等底层操作
 * 2. QueueMessage  - 管理单个队列的消息，包括内存状态和持久化操作
 * 3. MessageManager - 全局消息管理器，管理所有队列的消息对象
 */

#include "../common/logger.hpp"
#include "../common/helper.hpp"
#include "../common/msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <list>


namespace bitmq 
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"
    using MessagePtr = std::shared_ptr<bitmq::Message>;
    
    /**
     * @class MessageMapper
     * @brief 负责消息的持久化存储管理，处理文件读写、垃圾回收等底层操作
     * 
     * 管理特定队列的数据文件和临时文件，提供消息插入、删除和垃圾回收功能。
     */
    class MessageMapper 
    {
        public:
            /**
             * @brief 构造函数，初始化队列数据文件路径
             * @param basedir 队列数据存储基础目录
             * @param qname 队列名称
             */
            MessageMapper(std::string &basedir, const std::string &qname):
                _qname(qname) {
                if (basedir.back() != '/') basedir.push_back('/');
                _datafile = basedir + qname + DATAFILE_SUBFIX;
                _tmpfile = basedir + qname + TMPFILE_SUBFIX;
                if (FileHelper(basedir).exists() == false) 
                {
                    assert(FileHelper::createDirectory(basedir));
                }
                createMsgFile();
            }
            
            /**
             * @brief 创建队列数据文件
             * @return 创建成功返回true，否则返回false
             */
            bool createMsgFile() 
            {
                if (FileHelper(_datafile).exists() == true) 
                {
                    return true;
                }
                bool ret = FileHelper::createFile(_datafile);
                if (ret == false) 
                {
                    DLOG("创建队列数据文件 %s 失败！", _datafile.c_str());
                    return false;
                }
                return true;
            }
            
            /**
             * @brief 删除队列数据文件和临时文件
             */
            void removeMsgFile() 
            {
                FileHelper::removeFile(_datafile);
                FileHelper::removeFile(_tmpfile);
            }
            
            /**
             * @brief 插入消息到数据文件
             * @param msg 待插入的消息指针
             * @return 插入成功返回true，否则返回false
             */
            bool insert(MessagePtr &msg) 
            {
                return insert(_datafile, msg);
            }
            
            /**
             * @brief 逻辑删除消息（标记为无效）
             * @param msg 待删除的消息指针
             * @return 删除成功返回true，否则返回false
             */
            bool remove(MessagePtr &msg) 
            {
                //1. 将msg中的有效标志位修改为 '0'
                msg->mutable_payload()->set_valid("0");
                //2. 对msg进行序列化
                std::string body = msg->payload().SerializeAsString();
                if (body.size() != msg->length()) 
                {
                    DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致!");
                    return false;
                }
                //3. 将序列化后的消息，写入到数据在文件中的指定位置（覆盖原有的数据）
                FileHelper helper(_datafile);
                bool ret = helper.write(body.c_str(), msg->offset(), body.size());
                if (ret == false) 
                {
                    DLOG("向队列数据文件写入数据失败！");
                    return false;
                }
                //DLOG("确认消息后，删除持久化消息成功：%s", msg->payload().body().c_str());
                return true;
            }
            
            /**
             * @brief 执行垃圾回收，整理有效消息
             * @return 垃圾回收后有效的消息列表
             */
            std::list<MessagePtr> gc() 
            {
                bool ret;
                std::list<MessagePtr> result;
                ret = load(result);
                if (ret == false) 
                {
                    DLOG("加载有效数据失败！\n");
                    return result;
                }
                //DLOG("垃圾回收，得到有效消息数量：%d", result.size());
                //2. 将有效数据，进行序列化存储到临时文件中
                FileHelper::createFile(_tmpfile);
                for (auto &msg : result) 
                {
                    DLOG("向临时文件写入数据: %s", msg->payload().body().c_str());
                    ret = insert(_tmpfile, msg);
                    if (ret == false) 
                    {
                        DLOG("向临时文件写入消息数据失败！！");
                        return result;
                    }
                }
                //DLOG("垃圾回收后，向临时文件写入数据完毕，临时文件大小: %ld", FileHelper(_tmpfile).size());
                //3. 删除源文件
                ret = FileHelper::removeFile(_datafile);
                if (ret == false) 
                {
                    DLOG("删除源文件失败！");
                    return result;
                }
                //4. 修改临时文件名，为源文件名称
                ret = FileHelper(_tmpfile).rename(_datafile);
                if (ret == false) 
                {
                    DLOG("修改临时文件名称失败！");
                    return result;
                }
                //5. 返回新的有效数据
                return result;
            }
        private:
            /**
             * @brief 从数据文件加载所有有效消息
             * @param result 输出参数，存储加载的有效消息
             * @return 加载成功返回true，否则返回false
             */
            bool load(std::list<MessagePtr> &result) 
            {
                //1. 加载出文件中所有的有效数据；  存储格式 4字节长度|数据|4字节长度|数据.....
                FileHelper data_file_helper(_datafile);
                size_t offset = 0,msg_size;
                size_t fsize = data_file_helper.size();
                //DLOG("准备开始加载持久化数据，当前文件大小: %ld", data_file_helper.size());
                bool ret;
                while(offset < fsize) 
                {
                    ret = data_file_helper.read((char*)&msg_size, offset, sizeof(size_t));
                    if (ret == false) 
                    {
                        DLOG("读取消息长度失败！");
                        return false;
                    }
                    offset += sizeof(size_t);
                    std::string msg_body(msg_size, '\0');
                    data_file_helper.read(&msg_body[0], offset, msg_size);
                    if (ret == false) 
                    {
                        DLOG("读取消息数据失败！");
                        return false;
                    }
                    offset += msg_size;
                    MessagePtr msgp = std::make_shared<Message>();
                    msgp->mutable_payload()->ParseFromString(msg_body);
                    //DLOG("加载到有效数据：%s", msgp->payload().body().c_str());
                    //如果是无效消息，则直接处理下一个
                    if (msgp->payload().valid() == "0")  
                    {
                        DLOG("加载到无效消息：%s", msgp->payload().body().c_str());
                        continue;
                    }
                    //有效消息则保存起来
                    result.push_back(msgp);
                }
                return true;
            }
            
            /**
             * @brief 向指定文件插入消息
             * @param filename 目标文件名
             * @param msg 待插入的消息指针
             * @return 插入成功返回true，否则返回false
             */
            bool insert(const std::string &filename, MessagePtr &msg) 
            {
                //新增数据都是添加在文件末尾的
                //1. 进行消息的序列化，获取到格式化后的消息
                std::string body = msg->payload().SerializeAsString();
                //2. 获取文件长度
                FileHelper helper(filename);
                size_t fsize = helper.size();
                size_t msg_size = body.size();
                //写入逻辑：1. 先写入4字节数据长度， 2， 再写入指定长度数据
                bool ret = helper.write((char*)&msg_size, fsize, sizeof(size_t));
                if (ret == false) 
                {
                    DLOG("向队列数据文件写入数据长度失败！");
                    return false;
                }
                //3. 将数据写入文件的指定位置
                ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
                if (ret == false)
                {
                    DLOG("向队列数据文件写入数据失败！");
                    return false;
                }
                //4. 更新msg中的实际存储信息
                msg->set_offset(fsize + sizeof(size_t));
                msg->set_length(body.size());
                return true;
            }
        private:
            std::string _qname;       ///< 队列名称
            std::string _datafile;    ///< 队列数据文件路径
            std::string _tmpfile;     ///< 临时文件路径
    };

    /**
     * @class QueueMessage
     * @brief 管理单个队列的消息，包括内存状态和持久化操作
     * 
     * 负责消息的插入、获取、确认删除，以及与持久化层的交互。
     */
    class QueueMessage
    {
        public:
            using ptr = std::shared_ptr<QueueMessage>;
            
            /**
             * @brief 构造函数
             * @param basedir 数据存储基础目录
             * @param qname 队列名称
             */
            QueueMessage(std::string &basedir, const std::string &qname):_mapper(basedir, qname), 
                _qname(qname), _valid_count(0), _total_count(0) {}
            
            /**
             * @brief 从持久化存储中恢复消息
             * @return 恢复成功返回true
             */
            bool recovery() 
            {
                //恢复历史消息
                std::unique_lock<std::mutex> lock(_mutex);
                _msgs = _mapper.gc();
                for (auto &msg : _msgs) 
                {
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                }
                _valid_count = _total_count = _msgs.size();
                return true;
            }
            
            /**
             * @brief 插入新消息到队列
             * @param bp 消息属性（可为nullptr）
             * @param body 消息体
             * @param queue_is_durable 队列是否持久化
             * @return 插入成功返回true，否则返回false
             */
            bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable) 
            {
                //1. 构造消息对象
                MessagePtr msg = std::make_shared<Message>();
                msg->mutable_payload()->set_body(body);
                if (bp != nullptr) 
                {
                    DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                    msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                    msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                    msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
                }
                else 
                {
                    DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                    msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                    msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                    msg->mutable_payload()->mutable_properties()->set_routing_key("");
                }
                std::unique_lock<std::mutex> lock(_mutex);
                //2. 判断消息是否需要持久化
                if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE) 
                {
                    msg->mutable_payload()->set_valid("1");//在持久化存储中表示数据有效
                    //3. 进行持久化存储
                    bool ret = _mapper.insert(msg);
                    if (ret == false) 
                    {
                        DLOG("持久化存储消息：%s 失败了！", body.c_str());
                        return false;
                    }
                    _valid_count += 1; //持久化信息中的数据量+1
                    _total_count += 1;
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                }
                //4. 内存的管理
                _msgs.push_back(msg);
                return true;
            }
            
            /**
             * @brief 获取队列首条消息
             * @return 消息指针，队列为空时返回空指针
             */
            MessagePtr front()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if (_msgs.size() == 0) 
                {
                    return MessagePtr();
                }
                //获取一条队首消息：从_msgs中取出数据
                MessagePtr msg = _msgs.front();
                _msgs.pop_front();
                //将该消息对象，向待确认的hash表中添加一份，等到收到消息确认后进行删除
                _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                return msg;
            }
            
            /**
             * @brief 确认删除消息
             * @param msg_id 要删除的消息ID
             * @return 操作成功返回true
             * 
             * 每次删除后判断是否需要触发垃圾回收
             */
            bool remove(const std::string &msg_id) 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //1. 从待确认队列中查找消息
                auto it = _waitack_msgs.find(msg_id);
                if (it == _waitack_msgs.end()) 
                {
                    DLOG("没有找到要删除的消息：%s!", msg_id.c_str());
                    return true;
                }
                //2. 根据消息的持久化模式，决定是否删除持久化信息
                if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE) 
                {
                    //3. 删除持久化信息
                    _mapper.remove(it->second);
                    _durable_msgs.erase(msg_id);
                    _valid_count -= 1;//持久化文件中有效消息数量 -1
                    gc();  //内部判断是否需要垃圾回收，需要的话则回收一下
                }
                //4. 删除内存中的信息
                _waitack_msgs.erase(msg_id);
                //DLOG("确认消息后，删除消息的管理成功：%s", it->second->payload().body().c_str());
                return true;
            }
            
            /**
             * @brief 获取可获取的消息数量（待推送消息数）
             * @return 消息数量
             */
            size_t getable_count() 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _msgs.size();
            }
            
            /**
             * @brief 获取总消息数量（包含待确认消息）
             * @return 消息总数量
             */
            size_t total_count() 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _total_count;
            }
            
            /**
             * @brief 获取持久化消息数量
             * @return 持久化消息数量
             */
            size_t durable_count() 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _durable_msgs.size();
            }
            
            /**
             * @brief 获取待确认消息数量
             * @return 待确认消息数量
             */
            size_t waitack_count() 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                return _waitack_msgs.size();
            }
            
            /**
             * @brief 清空队列所有消息
             */
            void clear() 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _mapper.removeMsgFile();
                _msgs.clear();
                _durable_msgs.clear();
                _waitack_msgs.clear();
                _valid_count = 0;
                _total_count = 0;
            }
        private:
            /**
             * @brief 检查是否需要执行垃圾回收
             * @return 需要回收返回true，否则返回false
             */
            bool GCCheck() 
            {
                //持久化的消息总量大于2000， 且其中有效比例低于50%则需要持久化
                if (_total_count > 2000 && _valid_count * 10  / _total_count < 5) {
                    return true;
                }
                return false;
            }
            
            /**
             * @brief 执行垃圾回收，整理持久化文件
             */
            void gc() 
            {
                //1. 进行垃圾回收，获取到垃圾回收后，有效的消息信息链表
                if (GCCheck() == false) return;
                std::list<MessagePtr> msgs = _mapper.gc();
                for (auto &msg : msgs) 
                {
                    auto it = _durable_msgs.find(msg->payload().properties().id());
                    if (it == _durable_msgs.end()) 
                    {
                        DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理!");
                        _msgs.push_back(msg); ///做法：重新添加到推送链表的末尾
                        _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                        continue;
                    }
                    //2. 更新每一条消息的实际存储位置
                    it->second->set_offset(msg->offset());
                    it->second->set_length(msg->length());
                }
                //3. 更新当前的有效消息数量 & 总的持久化消息数量
                _valid_count = _total_count = msgs.size();
            }
        private:
            std::mutex _mutex;                           ///< 互斥锁，保证线程安全
            std::string _qname;                          ///< 队列名称
            size_t _valid_count;                         ///< 有效持久化消息计数
            size_t _total_count;                         ///< 总持久化消息计数
            MessageMapper _mapper;                       ///< 持久化映射器
            std::list<MessagePtr> _msgs;                 ///< 待推送消息列表
            std::unordered_map<std::string, MessagePtr> _durable_msgs; ///< 持久化消息映射表
            std::unordered_map<std::string, MessagePtr> _waitack_msgs; ///< 待确认消息映射表
    };

    /**
     * @class MessageManager
     * @brief 全局消息管理器，管理所有队列的消息对象
     * 
     * 提供队列的创建、销毁及消息操作接口，作为消息队列系统的核心管理类。
     */
    class MessageManager 
    {
        public:
            using ptr = std::shared_ptr<MessageManager>;
            
            /**
             * @brief 构造函数
             * @param basedir 数据存储基础目录
             */
            MessageManager(const std::string &basedir): _basedir(basedir){}
            
            /**
             * @brief 清空所有队列消息
             */
            void clear() 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                for (auto &qmsg : _queue_msgs) 
                {
                    qmsg.second->clear();
                }
            }
            
            /**
             * @brief 初始化队列消息对象
             * @param qname 队列名称
             */
            void initQueueMessage(const std::string &qname) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it != _queue_msgs.end()) 
                    {
                        return ;
                    }
                    qmp = std::make_shared<QueueMessage>(_basedir, qname);
                    _queue_msgs.insert(std::make_pair(qname, qmp));
                }
                qmp->recovery();
            }
            
            /**
             * @brief 销毁队列消息对象
             * @param qname 队列名称
             */
            void destroyQueueMessage(const std::string &qname) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) 
                    {
                        return ;
                    }
                    qmp = it->second;
                    _queue_msgs.erase(it);
                }
                qmp->clear();
            }
            
            /**
             * @brief 向指定队列插入消息
             * @param qname 队列名称
             * @param bp 消息属性
             * @param body 消息体
             * @param queue_is_durable 队列是否持久化
             * @return 插入成功返回true，否则返回false
             */
            bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable) {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) 
                    {
                        DLOG("向队列%s新增消息失败：没有找到消息管理句柄!", qname.c_str());
                        return false;
                    }
                    qmp = it->second;
                }
                return qmp->insert(bp, body, queue_is_durable);
            }
            
            /**
             * @brief 获取队列首条消息
             * @param qname 队列名称
             * @return 消息指针，队列为空或未找到返回空指针
             */
            MessagePtr front(const std::string &qname) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) {
                        DLOG("获取队列%s队首消息失败：没有找到消息管理句柄!", qname.c_str());
                        return MessagePtr();
                    }
                    qmp = it->second;
                }
                return qmp->front();
            }
            
            /**
             * @brief 确认删除队列中的消息
             * @param qname 队列名称
             * @param msg_id 消息ID
             */
            void ack(const std::string &qname, const std::string &msg_id) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) 
                    {
                        DLOG("确认队列%s消息%s失败：没有找到消息管理句柄!", qname.c_str(), msg_id.c_str());
                        return ;
                    }
                    qmp = it->second;
                }
                qmp->remove(msg_id);
                return ;
            }
            
            /**
             * @brief 获取队列可获取消息数量
             * @param qname 队列名称
             * @return 可获取消息数量
             */
            size_t getable_count(const std::string &qname) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) 
                    {
                        DLOG("获取队列%s待推送消息数量失败：没有找到消息管理句柄!", qname.c_str());
                        return 0;
                    }
                    qmp = it->second;
                }
                return qmp->getable_count();
            }
            
            /**
             * @brief 获取队列总消息数量
             * @param qname 队列名称
             * @return 总消息数量
             */
            size_t total_count(const std::string &qname) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) 
                    {
                        DLOG("获取队列%s总持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                        return 0;
                    }
                    qmp = it->second;
                }
                return qmp->total_count();
            }
            
            /**
             * @brief 获取队列持久化消息数量
             * @param qname 队列名称
             * @return 持久化消息数量
             */
            size_t durable_count(const std::string &qname) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) 
                    {
                        DLOG("获取队列%s有效持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                        return 0;
                    }
                    qmp = it->second;
                }
                return qmp->durable_count();
            }
            
            /**
             * @brief 获取队列待确认消息数量
             * @param qname 队列名称
             * @return 待确认消息数量
             */
            size_t waitack_count(const std::string &qname) 
            {
                QueueMessage::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _queue_msgs.find(qname);
                    if (it == _queue_msgs.end()) 
                    {
                        DLOG("获取队列%s待确认消息数量失败：没有找到消息管理句柄!", qname.c_str());
                        return 0;
                    }
                    qmp = it->second;
                }
                return qmp->waitack_count();
            }
        private:
            std::mutex _mutex;  ///< 互斥锁，保证线程安全
            std::string _basedir; ///< 数据存储基础目录
            std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; ///< 队列消息对象映射表
    };
}
#endif 